package com.example.calcite.csv;

import com.example.calcite.ITable;
import org.apache.calcite.DataContext;
import org.apache.calcite.linq4j.Enumerable;
import org.apache.calcite.linq4j.Linq4j;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.schema.impl.AbstractTable;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;

import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class CsvTable extends AbstractTable implements ITable {
    private final String filePath;
    private String[] columns;
    private List<Object[]> rows;

    public CsvTable(String filePath) throws IOException {
        this.filePath = filePath;
        this.rows = new ArrayList<>();
        loadData();
    }

    private void loadData() throws IOException {
        try (FileReader reader = new FileReader(filePath);
             CSVParser parser = new CSVParser(reader,
                     CSVFormat.DEFAULT.builder()
                             .setHeader()
                             .setSkipHeaderRecord(true)
                             .build())) {

            // Get column names from header
            columns = parser.getHeaderNames().toArray(new String[0]);

            // Read data rows
            for (CSVRecord record : parser) {
                Object[] row = new Object[columns.length];
                for (int i = 0; i < columns.length; i++) {
                    String value = record.get(i);
                    // Try to convert to appropriate type
                    try {
                        if (value.matches("-?\\d+")) {
                            row[i] = Integer.parseInt(value);
                        } else if (value.matches("-?\\d+\\.\\d+")) {
                            row[i] = Double.parseDouble(value);
                        } else {
                            row[i] = value;
                        }
                    } catch (NumberFormatException e) {
                        row[i] = value;
                    }
                }
                rows.add(row);
            }
        }
    }

    @Override
    public String[] getColumns() {
        return columns;
    }


    @Override
    public void insert(Object... values) {
        throw new UnsupportedOperationException("Insert operation is not supported for CSV tables");
    }

    @Override
    public RelDataType getRowType(RelDataTypeFactory typeFactory) {
        RelDataTypeFactory.Builder builder = typeFactory.builder();
        for (String column : columns) {
            if (column.equalsIgnoreCase("ID")) {
                builder.add(column, typeFactory.createSqlType(SqlTypeName.INTEGER));
            } else if (column.equalsIgnoreCase("AGE")) {
                builder.add(column, typeFactory.createSqlType(SqlTypeName.INTEGER));
            } else if (column.equalsIgnoreCase("SALARY")) {
                builder.add(column, typeFactory.createSqlType(SqlTypeName.DECIMAL));
            } else {
                builder.add(column, typeFactory.createSqlType(SqlTypeName.VARCHAR));
            }
        }
        return builder.build();
    }

    /*@Override
    public Enumerable<Object[]> scan(DataContext root) {
        return Linq4j.asEnumerable(rows);
    }*/

    @Override
    public Enumerable<Object[]> scan(DataContext root, List<RexNode> filters) {
        return Linq4j.asEnumerable(rows);
    }
}